package beautiful.butterfly.drds.data_exchange.core.taskgroup.runner;


import beautiful.butterfly.drds.data_exchange.core.statistics.communication.CommunicationTool;
import common.plugin.AbstractTaskPlugin;
import common.plugin.Consume;
import common.spi.Writer;
import common.statistics.PerfRecord;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 单个slice的writer执行调用
 */
public class WriterRunner extends AbstractRunner implements Runnable
{

    private static final Logger LOG = LoggerFactory
            .getLogger(WriterRunner.class);

    private Consume consume;

    public WriterRunner(AbstractTaskPlugin abstractTaskPlugin)
    {
        super(abstractTaskPlugin);
    }

    public void setConsume(Consume consume)
    {
        this.consume = consume;
    }

    //@Override
    public void run()
    {
        Validate.isTrue(this.consume != null);

        Writer.Task taskWriter = (Writer.Task) this.getPlugin();
        //统计waitReadTime，并且在finally end
        PerfRecord channelWaitRead = new PerfRecord(getTaskGroupId(), getTaskId(), PerfRecord.PHASE.WAIT_READ_TIME);
        try
        {
            channelWaitRead.start();
            LOG.debug("task writer starts to do init ...");
            PerfRecord initPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), PerfRecord.PHASE.WRITE_TASK_INIT);
            initPerfRecord.start();
            taskWriter.init();
            initPerfRecord.end();

            LOG.debug("task writer starts to do prepare ...");
            PerfRecord preparePerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), PerfRecord.PHASE.WRITE_TASK_PREPARE);
            preparePerfRecord.start();
            taskWriter.prepare();
            preparePerfRecord.end();
            LOG.debug("task writer starts to write ...");

            PerfRecord dataPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), PerfRecord.PHASE.WRITE_TASK_DATA);
            dataPerfRecord.start();
            taskWriter.startWrite(consume);

            dataPerfRecord.addCount(CommunicationTool.getTotalReadRecords(super.getRunnerCommunication()));
            dataPerfRecord.addSize(CommunicationTool.getTotalReadBytes(super.getRunnerCommunication()));
            dataPerfRecord.end();

            LOG.debug("task writer starts to do post ...");
            PerfRecord postPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), PerfRecord.PHASE.WRITE_TASK_POST);
            postPerfRecord.start();
            taskWriter.post();
            postPerfRecord.end();

            super.markSuccess();
        } catch (Throwable e)
        {
            LOG.error("Writer Runner Received Exceptions:", e);
            super.markFail(e);
        } finally
        {
            LOG.debug("task writer starts to do destroy ...");
            PerfRecord desPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), PerfRecord.PHASE.WRITE_TASK_DESTROY);
            desPerfRecord.start();
            super.destroy();
            desPerfRecord.end();
            channelWaitRead.end(super.getRunnerCommunication().getLongCounter(CommunicationTool.wait_reader_time));
        }
    }

    public boolean supportFailOver()
    {
        Writer.Task taskWriter = (Writer.Task) this.getPlugin();
        return taskWriter.supportFailOver();
    }

    public void shutdown()
    {
        consume.shutdown();
    }
}
